package com.microcity.rxjava2exe;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * RxJava2  线程控制  1.读写数据库 2.网络请求  在博客的Demo中
 * <p>
 * 在RxJava中, 已经内置了很多线程选项供我们选择, 例如有
 * Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
 * Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
 * Schedulers.newThread() 代表一个常规的新线程
 * AndroidSchedulers.mainThread() 代表Android的主线程
 * <p>
 * <p>
 * 如果在请求的过程中Activity已经退出了, 这个时候如果回到主线程去更新UI, 那么APP肯定就崩溃了, 怎么办呢,
 * 上一节我们说到了Disposable , 说它是个开关, 调用它的dispose()方法时就会切断水管, 使得下游收不到事件,
 * 既然收不到事件, 那么也就不会再去更新UI了. 因此我们可以在Activity中将这个Disposable 保存起来,
 * 当Activity退出时, 切断它即可.
 * <p>
 * 那如果有多个Disposable 该怎么办呢, RxJava中已经内置了一个容器CompositeDisposable,
 * 每当我们得到一个Disposable时就调用CompositeDisposable.add()将它添加到容器中, 在退出的时候,
 * 调用CompositeDisposable.clear() 即可切断所有的水管.
 */
public class Exe2Activity extends AppCompatActivity {

    private String TAG = "MY" + this.getClass().getSimpleName();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_exe2);
        //上游在哪个线程发,下游在哪个线程接收(处理)
//        initRx1();
        //切换线程，上游在子线程或者IO线程，下游在主线程更新(处理)
//        initRx2();
//        initRx3();
        //再次验证initRx3()
        initRx4();

    }

    private void initRx4() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "Observable thread is " + Thread.currentThread().getName());
                Log.d(TAG, "emit " + 1);
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                Log.d(TAG, "onNext: " + integer);
            }
        };
        /**
         *1.多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
         *2.多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.
         */
        observable.subscribeOn(Schedulers.newThread())//上游在新的子线程 observable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())//下游在主线程中进行 consumer
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "observeOn(mainThread),currentThread is " + Thread.currentThread().getName());
                    }
                })
                .observeOn(Schedulers.io())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "observeOn(io),currentThread is " + Thread.currentThread().getName());
                    }
                })
                .subscribe(consumer);
    }

    private void initRx3() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "Observable thread is " + Thread.currentThread().getName());
                Log.d(TAG, "emit " + 1);
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                Log.d(TAG, "onNext: " + integer);
            }
        };
        /**
         *1.多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
         *2.多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.
         */
        observable.subscribeOn(Schedulers.newThread())//上游在新的子线程 observable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())//下游在主线程中进行 consumer
                .observeOn(Schedulers.io())
                .subscribe(consumer);
    }

    private void initRx1() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "Observable thread is " + Thread.currentThread().getName());
                Log.d(TAG, "emit " + 1);
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                Log.d(TAG, "onNext: " + integer);
            }
        };
        observable.subscribe(consumer);
    }

    private void initRx2() {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "Observable thread is " + Thread.currentThread().getName());
                Log.d(TAG, "emit " + 1);
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                Log.d(TAG, "onNext: " + integer);
            }
        };
        /**
         * subscribeOn() 指定的是上游发送事件的线程, observeOn() 指定的是下游接收事件的线程.
         */
        observable.subscribeOn(Schedulers.newThread())//上游在新的子线程 observable
                .observeOn(AndroidSchedulers.mainThread())//下游在主线程中进行 consumer
                .subscribe(consumer);
    }


}
